package com.cennavi.funcs;

import java.text.DecimalFormat;
import java.util.HashMap;
import java.util.Iterator;

import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;

import scala.Tuple2;

/** * @author  作者 :allen
    * @date 创建时间：2017年9月30日 下午2:11:40 
    * @version 1.0 
    */

public class FindMinFlowPairFunction1 implements PairFunction<Tuple2<String, Iterable<String>>, String, String> {

	private static final long serialVersionUID = 1L;
	private HashMap<String, String> road_capaMap;
	public FindMinFlowPairFunction1(Broadcast<HashMap<String, String>> road_capaBroadMap) {
		this.road_capaMap = road_capaBroadMap.value();
	}

	public Tuple2<String, String> call(Tuple2<String, Iterable<String>> arg0) throws Exception {
		String value = "";
		String[] data;
		int minFlow = 100000;
		String returnKey = arg0._1;
		String returnValue = "";
		for(Iterator<String> iter = arg0._2.iterator();iter.hasNext();){
			value = (String)iter.next();
			data = value.split(",");
			if(Integer.parseInt(data[9]) < minFlow){
				minFlow = Integer.parseInt(data[9]);
				returnValue = value;
			}
		}
		String capacity = "null";
		String saturation = "null";
		if(this.road_capaMap.containsKey(returnKey)){
			capacity = this.road_capaMap.get(returnKey);
			//一小时通行能力换算为15分钟
			capacity = String.valueOf((int)Math.ceil(Integer.parseInt(capacity) / 4.0));
		}
		if(!capacity.equals("null")){
			saturation = new DecimalFormat("######0.00").format(1.0 * minFlow / Integer.parseInt(capacity));
		}
		returnValue = returnValue.split(",")[0] + "," + 
						returnValue.split(",")[1] + "," + 
						returnValue.split(",")[2] + "," + 
						returnValue.split(",")[3] + "," + 
						returnValue.split(",")[4] + "," + 
						returnKey + "," + 
						returnValue.split(",")[6] + "," + 
						returnValue.split(",")[7] + "," + 
						String.valueOf(minFlow) + "," +
						saturation + "," + 
						capacity;
		Tuple2<String, String> returnData = new Tuple2<String, String>(returnKey, returnValue);
		return returnData;
	}

}
